package ru.mipt.victator.pagerank.finalization;

import java.io.IOException;
import java.io.PrintStream;
import java.util.Map;
import java.util.Properties;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;

import ru.mipt.victator.pagerank.rank.SequenceFile2Map;

public class Finalization
{
	private FileSystem FS;
	private JobConf conf;
	private Path pPRdir;
	private Path pLastPR;
	private Path pRes;
	private int last;
	private float StartPR;
	
	public void config(FileSystem _FS, Properties _props)
	{
		FS = _FS;
		conf = new JobConf();
		pPRdir = new Path(_props.getProperty("OutputData.PageRank.Dir"));
		pRes = new Path(pPRdir,"result");
		last = Integer.parseInt(_props.getProperty("RankJob.Iterations.Int"));
		pLastPR = new Path(pPRdir,Integer.toString(last));
		StartPR = Float.parseFloat(_props.getProperty("InputData.StartPageRank.Float"));
	}
	
	public void process() throws IOException
	{
		Map<String,FloatWritable> map = SequenceFile2Map.Read(FS,conf,pLastPR);
		
		FSDataOutputStream FSout = FS.create(pRes);
		PrintStream out = new PrintStream(FSout,true,"UTF-8");
		
		SequenceFile.Reader reader = new SequenceFile.Reader(FS, new Path(pPRdir,Integer.toString(0)),conf);
		boolean read = true;
		Text text = new Text();
		FloatWritable floatw = new FloatWritable();
		FloatWritable key;
		while(read)
		{
			read = reader.next(text, floatw);
			if (read)
			{
				key = map.get(text.toString());
				if (key != null) out.println(text + " = " + key.get());
				else out.println(text + " = " + StartPR);
			}
		}
		reader.close();
		out.close();
		System.out.println("Result was saved in " + pRes);
	}
}
